//package com.bs.config;
//
//import com.bs.sink.KafkaSink;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.connector.kafka.source.KafkaSource;
//import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.springframework.context.annotation.Configuration;
//
//import javax.annotation.PostConstruct;
//
///**
// * @author bingshao
// * @date 2022/11/11
// **/
//@Configuration
//public class CdcStreamKafkaToKafkaConfig {
//
//    @PostConstruct
//    public void initKafkaToKafkaFlink() throws Exception {
////        org.apache.flink.configuration.Configuration config = new org.apache.flink.configuration.Configuration();
////        config.setString("execution.savepoint.path", "C:\\checkpoint\\4955148a77700ae15ca9af388ad61cd3\\chk-5");
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.enableCheckpointing(60 * 1000L);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30 * 1000L);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 2 * 1000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
////        env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint");
//        KafkaSource<String> source = KafkaSource.<String>builder()
//                .setBootstrapServers("110.42.233.197:9092")
//                .setTopics("flink-test")
//                .setGroupId("flink.test.group")
//                .setStartingOffsets(OffsetsInitializer.latest())
//                .setValueOnlyDeserializer(new SimpleStringSchema())
//                .build();
//        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
//                .addSink(new KafkaSink()).setParallelism(1);
//        env.execute("Stream-Kafka-To-Kafka");
//    }
//}
